package com.atguigu.flink.sqlfunction;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Expressions;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

/**
 * Created by Smexy on 2023/3/5
 *
 *  flink内置的函数: https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/dev/table/functions/systemfunctions/
 */
public class Demo1_FunctionUse
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);

        String createTableSql = " create table t1 ( id string,ts bigint, vc int  )" +
            " with ( " +
            " 'connector' = 'filesystem' ,   " +
            " 'path' =  'data/cep.txt' ,   " +
            "  'format' = 'csv' " +
            "      )                 ";

        //执行建表
        tableEnvironment.executeSql(createTableSql);

        //TableAPI 用  Expresions.call(函数，参数)
        //从表名得到 Table
        Table table = tableEnvironment.from("t1");

        table.select($("id"),$("vc"),
                 Expressions.call("rand",$("vc")))
             .execute().print();


        //sql用
       /* tableEnvironment.sqlQuery("select id,vc,rand(vc) from t1")
                        .execute()
                        .print();*/

    }
}
